package com.suxin.notify;

import com.google.common.collect.Sets;
import com.suxin.notify.listener.Subscriber;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @author Tang
 * @classname DefaultPublisher
 * @description [  ]
 * @date 2022/9/30 14:58
 */
public class DefaultPublisher extends Thread implements EventPublisher {

    private volatile boolean shutdown = false;

    private volatile boolean initialized = false;

    protected final HashSet<Subscriber> subscribers = Sets.newHashSet();

    private BlockingQueue<Event> queue;
    private Class<? extends Event> eventType;

    @Override
    public void init(Class<? extends Event> type, int bufferSize) {
        setDaemon(true);
        setName("nacos.publisher-" + type.getName());
        this.queue = new ArrayBlockingQueue<>(bufferSize);
        this.eventType = type;
        start();
    }

    @Override
    public synchronized void start() {
        if (!initialized) {
            super.start();
            initialized = true;
        }
    }

    @Override
    public void run() {
        openEventHandler();
    }

    private void openEventHandler() {
        int waitTimes = 60;
        for (; ; ) {
            if (shutdown || !hasSubscribe() || waitTimes <= 0) {
                break;
            }
            try {
                TimeUnit.SECONDS.sleep(1000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            waitTimes--;
        }
        for (; ; ) {
            if (shutdown) {
                break;
            }
            Event event = null;
            try {
                event = queue.take();
            } catch (Exception e) {
                e.printStackTrace();
            }
            reviceEvent(event);
        }
    }

    private boolean hasSubscribe() {
        return CollectionUtils.isEmpty(subscribers);
    }

    @Override
    public void addSubscribe(Subscriber subscriber) {
        subscribers.add(subscriber);
    }

    @Override
    public void removeSubscribe(Subscriber subscriber) {

    }

    @Override
    public boolean publish(Event event) {
        boolean success = this.queue.offer(event);
        if (!success) {
            reviceEvent(event);
        }
        return true;
    }

    private void reviceEvent(Event event) {
        long sequence = event.getSequence();
        for (Subscriber subscriber : subscribers) {
            if (subscriber.ignoreExpireEvent()) {
                return;
            }
            notifySubscribe(subscriber,event);
        }
    }

    @Override
    public void notifySubscribe(Subscriber subscriber, Event event) {
        final Runnable job = () -> {
            subscriber.onEvent(event);
        };
        Executor executor = subscriber.executor();
        if (executor != null) {
            executor.execute(job);
        }else {
            job.run();
        }
    }

    @Override
    public void shutdown() throws Exception {
        shutdown = true;
    }
}
